package ex.datastream.windows;

import ex.datastream.ApiFrame;
import ex.datastream.functions.function.MyAggregateFunction;
import ex.datastream.functions.sourceFunction.SourceFuncByEvent;
import ex.vo.Event;
import org.apache.calcite.interpreter.Context;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.evictors.Evictor;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.runtime.operators.windowing.TimestampedValue;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.Optional;

/**
 * ref:https://blog.csdn.net/weixin_44760145/article/details/129252804
 */
public class EvictBeforeAndAfter {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.addSource(new SourceFuncByEvent())
                .keyBy(event -> 1)
                .windowAll(GlobalWindows.create())
//                 窗口 只 保存 20 条最近数据
                .trigger(
                        // 1min 触发一次
                        new Trigger<Event, GlobalWindow>() {
                            @Override
                            public TriggerResult onElement(Event element, long timestamp, GlobalWindow window, TriggerContext ctx) throws Exception {

                                ValueState<Boolean> firstSeen = ctx.getPartitionedState(new ValueStateDescriptor<Boolean>("first-seen", Types.BOOLEAN));
                                if (!Optional.ofNullable(firstSeen.value()).orElseGet(() -> false)) {
                                    // 窗口的第一个元素
                                    //注册一个 1min 的定时器
                                    long ts = ctx.getCurrentProcessingTime() + 5 * 1000L;
                                    SimpleDateFormat dateformat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                    System.out.println("【初始化定时器】 " + dateformat.format(ts));
                                    ctx.registerProcessingTimeTimer(ts);
                                    firstSeen.update(true);
                                }
                                System.out.printf("%s,",element.key);
                                return TriggerResult.CONTINUE;
                            }

                            @Override
                            public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
                                SimpleDateFormat dateformat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                System.out.println("\n【触发onProcessingTime】 = " +  dateformat.format(time));
                                ctx.registerProcessingTimeTimer(time +  5 * 1000L);
                                return TriggerResult.FIRE;
                            }

                            @Override
                            public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
                                return TriggerResult.CONTINUE;
                            }

                            @Override
                            public void clear(GlobalWindow window, TriggerContext ctx) throws Exception {
                                ValueState<Boolean> firstSeen = ctx.getPartitionedState(new ValueStateDescriptor<Boolean>("first-seen", Types.BOOLEAN));
                                firstSeen.clear();
                            }
                        }
                )

                .evictor(
                        // 保留最近20条数据
                        new Evictor<Event, GlobalWindow>() {
                            private int maxCount = 20;
                            @Override
                            public void evictBefore(Iterable<TimestampedValue<Event>> elements, int size, GlobalWindow window, EvictorContext evictorContext) {
                                SimpleDateFormat dateformat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                System.out.println("evictBefore: " + dateformat.format( evictorContext.getCurrentProcessingTime()));
                                long total = elements.spliterator().getExactSizeIfKnown();
                                if (total < maxCount) {
                                    return;
                                } else {
                                    // 需要移除窗口元素的数量 evictedCount = total - maxCount
                                    int evictedCount = 0;
                                    for (Iterator<TimestampedValue<Event>> iterator = elements.iterator(); iterator.hasNext(); ) {
                                        TimestampedValue<Event> element = iterator.next();
                                        evictedCount++;
                                        if (evictedCount > total - maxCount) {
                                            break;
                                        } else {
                                            System.out.printf(+element.getValue().key+",");
                                            iterator.remove();
                                        }
                                    }
                                    System.out.printf("<----删除条数:%s,总条数:%s,保留最近20条数据\n",evictedCount-1,total);
                                }
                            }
                            @Override
                            public void evictAfter(Iterable<TimestampedValue<Event>> elements, int size, GlobalWindow window, EvictorContext evictorContext) {
                                SimpleDateFormat dateformat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                System.out.println("evictAfter: " + dateformat.format( evictorContext.getCurrentProcessingTime()));
                            }
                        }
                )
                .process(
                        new ProcessAllWindowFunction<Event, String, GlobalWindow>() {
                            @Override
                            public void process(Context context, Iterable<Event> elements, Collector<String> out) throws Exception {
                                out.collect("窗口中有" + elements.spliterator().getExactSizeIfKnown() + "条元素");
                                System.out.println("****************************最新20条数据*****************************");
                                elements.forEach(a-> System.out.printf(a.key+","));
                                System.out.println("\n******************************************************************");
                            }
                        }
                )
                .print();

        env.execute();
    }
//【初始化定时器】 2024-02-29 15:51:06
//46,34,76,42,9,59,88,35,97,28,90,22,42,65,62,0,67,96,61,89,21,54,59,87,57,
//【触发onProcessingTime】 = 2024-02-29 15:51:06
//46,34,76,42,9,<----删除条数:5,总条数:25,保留最近20条数据
//窗口中有20条元素
//****************************最新20条数据*****************************
//59,88,35,97,28,90,22,42,65,62,0,67,96,61,89,21,54,59,87,57,
//******************************************************************
//
//82,91,81,57,96,37,98,13,56,67,22,73,36,49,13,40,50,80,75,25,8,47,5,48,
//【触发onProcessingTime】 = 2024-02-29 15:51:11
//59,88,35,97,28,90,22,42,65,62,0,67,96,61,89,21,54,59,87,57,82,91,81,57,<----删除条数:24,总条数:44,保留最近20条数据
//窗口中有20条元素
//****************************最新20条数据*****************************
//96,37,98,13,56,67,22,73,36,49,13,40,50,80,75,25,8,47,5,48,
//******************************************************************
//
//86,56,21,40,37,23,12,16,41,15,94,16,81,80,99,49,71,63,95,57,33,36,1,30,
//【触发onProcessingTime】 = 2024-02-29 15:51:16
//96,37,98,13,56,67,22,73,36,49,13,40,50,80,75,25,8,47,5,48,86,56,21,40,<----删除条数:24,总条数:44,保留最近20条数据
//窗口中有20条元素
//****************************最新20条数据*****************************
//37,23,12,16,41,15,94,16,81,80,99,49,71,63,95,57,33,36,1,30,
//******************************************************************
}

